// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: jshin

#pragma once

#include <cstdlib>
#include <functional>
#include <iostream>
#include <memory>
#include <string>
#include <utility>
#include <vector>

#include <boost/asio/connect.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/ssl/error.hpp>
#include <boost/asio/ssl/stream.hpp>
#include <boost/beast/core.hpp>
#include <boost/beast/http.hpp>
#include <boost/beast/version.hpp>
#include <coin/proto/coin_query.pb.h>

#include "coin2/base/log.h"
#include "coin2/base/request/session.h"

struct AsioHttpContext;

// TODO(jshin): this corrupts the namespace
enum AsioHttpContextStatus {
  kUnknown = 0,
  kCreated,
  kRequested,
  kResponded,
  kTimeout,
  kError,
  kCount
};

using AsioHttpContextWeight = coin2::base::request::RequestPriorityWeight;

struct AsioHttpContext {
  using HttpContextProto = coin::proto::HttpContextProto;

  static AsioHttpContext* NewGet(
      const std::string& host_,
      const std::string& port_,
      const std::string& bind_ip_,
      const std::string& target_) {
    return new AsioHttpContext(
        host_,
        port_,
        bind_ip_,
        target_,
        "",
        {},
        boost::beast::http::verb::get);
  }
  static AsioHttpContext* NewGet(
      const std::string& host_,
      const std::string& port_,
      const std::string& bind_ip_,
      const std::string& target_,
      const std::vector<std::pair<std::string, std::string>>& headers_) {
    LOG_IF(INFO, target_.size() > 3000) << target_;
    return new AsioHttpContext(
        host_,
        port_,
        bind_ip_,
        target_,
        "",
        headers_,
        boost::beast::http::verb::get);
  }
  static AsioHttpContext* NewPost(
      const std::string& host_,
      const std::string& port_,
      const std::string& bind_ip_,
      const std::string& target_) {
    return new AsioHttpContext(
        host_,
        port_,
        bind_ip_,
        target_,
        "",
        {},
        boost::beast::http::verb::post);
  }
  static AsioHttpContext* NewPost(
      const std::string& host_,
      const std::string& port_,
      const std::string& bind_ip_,
      const std::string& target_,
      const std::string& payload_) {
    LOG_IF(INFO, payload_.size() > 3000) << payload_;
    return new AsioHttpContext(
        host_,
        port_,
        bind_ip_,
        target_,
        payload_,
        {},
        boost::beast::http::verb::post);
  }
  static AsioHttpContext* NewPost(
      const std::string& host_,
      const std::string& port_,
      const std::string& bind_ip_,
      const std::string& target_,
      const std::vector<std::pair<std::string, std::string>>& headers_,
      const std::string& payload_) {
    LOG_IF(INFO, payload_.size() > 3000) << payload_;
    return new AsioHttpContext(
        host_,
        port_,
        bind_ip_,
        target_,
        payload_,
        headers_,
        boost::beast::http::verb::post);
  }
  static AsioHttpContext* NewDelete(
      const std::string& host_,
      const std::string& port_,
      const std::string& bind_ip_,
      const std::string& target_,
      const std::vector<std::pair<std::string, std::string>>& headers_,
      const std::string& payload_) {
    return new AsioHttpContext(
        host_,
        port_,
        bind_ip_,
        target_,
        payload_,
        headers_,
        boost::beast::http::verb::delete_);
  }
  static AsioHttpContext* NewDelete(
      const std::string& host_,
      const std::string& port_,
      const std::string& bind_ip_,
      const std::string& target_,
      const std::vector<std::pair<std::string, std::string>>& headers_) {
    return new AsioHttpContext(
        host_,
        port_,
        bind_ip_,
        target_,
        "",
        headers_,
        boost::beast::http::verb::delete_);
  }
  static AsioHttpContext* NewPut(
      const std::string& host_,
      const std::string& port_,
      const std::string& bind_ip_,
      const std::string& target_,
      const std::vector<std::pair<std::string, std::string>>& headers_,
      const std::string& payload_) {
    return new AsioHttpContext(
        host_,
        port_,
        bind_ip_,
        target_,
        payload_,
        headers_,
        boost::beast::http::verb::put);
  }

  AsioHttpContext(
      const std::string& host_,
      const std::string& port_,
      const std::string& bind_ip_,
      const std::string& target_,
      const std::string& payload_,
      const std::vector<std::pair<std::string, std::string>>& headers_,
      boost::beast::http::verb verb_)
      : host(host_),
        port(port_),
        bind_ip(bind_ip_),
        target(target_),
        payload(payload_),
        print_to_stdout(false),
        req{verb_, target, 11},

        status(AsioHttpContextStatus::kUnknown),
        weight(AsioHttpContextWeight::kMedium),
        status_ts{0},
        handler(nullptr) {
    req.set(boost::beast::http::field::host, host);
    req.set(boost::beast::http::field::user_agent, BOOST_BEAST_VERSION_STRING);
    bool header_has_content_type = false;
    for (const auto& pr : headers_) {
      req.set(pr.first, pr.second);
      if (!header_has_content_type && pr.first == "Content-Type") {
        header_has_content_type = true;
      }
    }
    if (!payload.empty()) {
      if (!header_has_content_type) {
        req.set(boost::beast::http::field::content_type, "application/json");
      }
      req.body() = payload;
      req.prepare_payload();
    }
  }

  void SetStatus(AsioHttpContextStatus status_, int64_t ts_) {
    status = status_;
    status_ts[static_cast<size_t>(status_)] = ts_;
  }

  void PrintLog() const {
    LOG(INFO) << "AsioHttpContext";
    LOG(INFO) << "  "
              << "url: " << host << " " << port << " " << target;
    LOG(INFO) << "  "
              << "status: " << status;
    LOG(INFO) << "  "
              << "status_ts: " << status_ts[1] << " " << status_ts[2] << " " << status_ts[3] << " "
              << status_ts[4] << " " << status_ts[5];
    LOG(INFO) << "  "
              << "req: " << req;
    LOG(INFO) << "  "
              << "res: " << res;
  }

  HttpContextProto ToProto(bool dump_complete_msg = true) const {
    HttpContextProto proto_msg;
    proto_msg.set_host(host);
    proto_msg.set_port(port);
    proto_msg.set_bind_ip(bind_ip);
    proto_msg.set_target(target);
    proto_msg.set_status(status);
    proto_msg.set_keep_user_agent(keep_user_agent);
    proto_msg.set_extra_info(extra_info);
    *proto_msg.mutable_error_code() = error_code;
    proto_msg.set_query_type(query_type);

    // setup response fields
    {
      auto* resp = proto_msg.mutable_http_response();
      resp->set_timestamp(response_timestamp);
      resp->set_status_code(res.result_int());
      auto& headers = *resp->mutable_headers();
      for (const auto& field : res.base()) {
        const auto key = std::string(field.name_string().data(), field.name_string().size());
        headers[key] = std::string(field.value().data(), field.value().size());
      }
      resp->set_body(res.body().substr(0, (dump_complete_msg ? std::string::npos : 1024)));
    }

    // setup request fields
    {
      auto* reqt = proto_msg.mutable_http_request();
      reqt->set_timestamp(request_timestamp);
      reqt->set_method(static_cast<coin::proto::HttpRequest_Method>(req.method()));
      reqt->set_version(req.version());

      const auto& t = req.target();
      reqt->set_target(std::string(t.data(), t.size()));
      auto& headers = *reqt->mutable_headers();
      for (const auto& field : req.base()) {
        const auto key = std::string(field.name_string().data(), field.name_string().size());
        headers[key] = std::string(field.value().data(), field.value().size());
      }
      reqt->set_body(req.body());
    }

    return proto_msg;
  }

  HttpContextProto ToProtoShort() const { return ToProto(false); }

  static std::shared_ptr<AsioHttpContext> FromProto(const HttpContextProto& proto_msg) {
    std::vector<std::pair<std::string, std::string>> headers;
    for (const auto& kv : proto_msg.http_request().headers()) {
      headers.emplace_back(kv);
    }
    auto method = static_cast<boost::beast::http::verb>(proto_msg.http_request().method());
    auto context = std::make_shared<AsioHttpContext>(
        proto_msg.host(),
        proto_msg.port(),
        proto_msg.bind_ip(),
        proto_msg.target(),
        proto_msg.http_request().body(),
        headers,
        method);

    context->status = static_cast<AsioHttpContextStatus>(proto_msg.status());
    context->keep_user_agent = proto_msg.keep_user_agent();
    context->extra_info = proto_msg.extra_info();
    context->error_code = proto_msg.error_code();
    context->query_type = proto_msg.query_type();

    context->res.version(11);
    context->res.result(proto_msg.http_response().status_code());
    for (const auto& kv : proto_msg.http_response().headers()) {
      context->res.set(kv.first, kv.second);
    }
    context->res.body() = proto_msg.http_response().body();
    return context;
  }

  const std::string host;
  const std::string port;
  const std::string bind_ip;
  const std::string target;
  const std::string payload;
  const bool print_to_stdout;
  int64_t response_timestamp{0};
  int64_t request_timestamp{0};
  boost::beast::http::request<boost::beast::http::string_body> req;
  boost::beast::http::response<boost::beast::http::string_body> res;

  AsioHttpContextStatus status;
  AsioHttpContextWeight weight;
  int64_t status_ts[static_cast<size_t>(AsioHttpContextStatus::kCount)];

  std::function<void(std::shared_ptr<AsioHttpContext>)> handler;
  bool keep_user_agent{false};
  std::string extra_info;
  int query_type{0};  //
  coin::proto::ErrorCode error_code;

  int64_t context_id = 0;
};
